package com.yuke.cloud.service.tmc.config;

import com.rabbitmq.client.Channel;
import com.yuke.cloud.common.base.constant.GlobalConstant;
import com.yuke.cloud.common.base.constant.MQConstants;
import com.yuke.cloud.common.config.YukeCoreConfig;
import com.yuke.cloud.common.core.config.RedisConfiguration;
import com.yuke.cloud.service.tmc.entity.TmcDeadLetterMsg;
import com.yuke.cloud.service.tmc.service.TmcDeadLetterMsgService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.apache.commons.lang3.StringUtils;

import javax.annotation.Resource;


/**
 * <p><b>Description:</b> RabbitMQ交换机、队列的配置类.定义交换机、key、queue并做好绑定。
 * 同时定义每个队列的ttl，队列最大长度，Qos等等
 * 这里只绑定了死信队列。建议每个队列定义自己的QueueConfig
 * <p><b>Company:</b> 
 *
 */
@Configuration
@ComponentScan(basePackageClasses={RedisConfiguration.class, YukeCoreConfig.class})  //用于rabbitmq相关
public class DeadQueueConfig {
	
	@Component
	public class DeadLetterMessageListener implements ChannelAwareMessageListener {
		private Logger logger = LoggerFactory.getLogger(DeadLetterMessageListener.class);

		@Autowired
		private RedisTemplate<Object, Object> redisTemplate;

		@Resource
		private TmcDeadLetterMsgService tmcDeadLetterMsgService;

		/*@Autowired
		private MailServiceImpl mailService;*/
		
		// 收件人
		/*@Value("${recipient.email.address}")
		private String emailRecipient;*/

		/**
		 * Callback for processing a received Rabbit message.
		 * <p>Implementors are supposed to process the given Message,
		 * typically sending reply messages through the given Session.
		 * @param message the received AMQP message (never <code>null</code>)
		 * @param channel the underlying Rabbit Channel (never <code>null</code>)
		 * @throws Exception Any.
		 */
		@Override
		public void onMessage(Message message, Channel channel) throws Exception {
			MessageProperties messageProperties = message.getMessageProperties();
	        // 消息体
			String messageBody = new String(message.getBody());

			logger.warn("dead letter message：{} | tag：{}", messageBody, message.getMessageProperties().getDeliveryTag());
			// 入库
			String logKey = messageProperties.getMessageId();
			insertRecord(logKey, message);
			/*// 发邮件
			sendEmail(logKey, messageProperties.getMessageId(), messageBody);*/

			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

			redisTemplate.opsForHash().delete(MQConstants.MQ_CONSUMER_RETRY_COUNT_KEY, messageProperties.getMessageId());
		}

		/**
		 * 入库，对异常无法处理的事务（死信）进行本地持久化存储，为后续手动处理作准备
		 */
		private void insertRecord(String logKey, Message message) {
			try {
				MessageProperties msgProp = message.getMessageProperties();
				TmcDeadLetterMsg tmcDeadLetterMsg = new TmcDeadLetterMsg();

				tmcDeadLetterMsg.setMsgId(msgProp.getMessageId());
				tmcDeadLetterMsg.setMsgBody(new String(message.getBody()));
				tmcDeadLetterMsg.setDeliverTag("" + msgProp.getDeliveryTag());
				tmcDeadLetterMsg.setMsgType(StringUtils.isNotBlank(msgProp.getType()) ?
						GlobalConstant.MQ_TYPE_PRODUCER : GlobalConstant.MQ_TYPE_CONSUMER);
				tmcDeadLetterMsg.setCreateTime(msgProp.getTimestamp());
				tmcDeadLetterMsg.setMsgExchange(msgProp.getReceivedExchange());
				tmcDeadLetterMsg.setMsgRoutingKey(msgProp.getReceivedRoutingKey());

				tmcDeadLetterMsgService.saveDeadMsg(tmcDeadLetterMsg);

				logger.info("{}|死信入库, {}", logKey, tmcDeadLetterMsg.toString());
			} catch (Exception e) {
				logger.error("{}|死信入库发生异常,{}", logKey, e.getMessage());
			}
		}

		/**
		 * 发邮件，当事务异常（死信）无法通过程序处理时，可通过邮件通知运维人员手动处理
		 *//*
		private void sendEmail(String logKey, String msgId, String msgBody) {
			try {
				String subject = "MQ处理异常";
				StringBuilder cnt = new StringBuilder();
				cnt.append("ID:" + msgId + "\n");
				cnt.append("消息体:" + msgBody + "\n");
				cnt.append("出现异常,请及时处理");
				mailService.sendSimpleMail(emailRecipient, subject, cnt.toString());
				logger.info("{}|发送邮件成功!", logKey);
			} catch (Exception e) {
				logger.error("{}|发送邮件----异常----", logKey, e.getMessage());
			}
		}*/
	}

    //========================== 声明交换机 ==========================
    /**
     * 死信交换机
     */
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(MQConstants.DLX_EXCHANGE);
    }

 

    //========================== 声明队列 ===========================
    /**
     * 死信队列
     */
    @Bean
    public Queue dlxQueue() {
        return new Queue(MQConstants.DLX_QUEUE,true,false,false);
    }
    /**
     * 通过死信路由key绑定死信交换机和死信队列
     */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange())
                .with(MQConstants.DLX_ROUTING_KEY);
    }
    
    /**
     * 死信队列的监听
     * @param connectionFactory RabbitMQ连接工厂
     * @param deadLetterMessageListener 死信的监听者
     * @return 监听容器对象
     */
    @Bean
    public SimpleMessageListenerContainer deadLetterListenerContainer(ConnectionFactory connectionFactory, 
    		DeadLetterMessageListener deadLetterMessageListener) {
    	
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(dlxQueue());
        container.setExposeListenerChannel(true);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(deadLetterMessageListener);
        /** 设置消费者能处理消息的最大个数 */
        container.setPrefetchCount(100);
        return container;
    }

    //====================== 一个例子，用来说明如何声明队列与交换机绑定 =======================
      
}
